package com.xya.rpc.client;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * RPC模式队列-客户端
 */
public class RpcClient implements AutoCloseable {


    private Connection connection;

    private Channel channel;

    //队列名称
    private static final String RPC_QUEUE_NAME = "rpc_queue";

    //初始化连接
    public RpcClient() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("xya");
        factory.setPassword("xya871219");
        factory.setVirtualHost("/xya");
        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] args) {
        try (RpcClient fibonacciRpc = new RpcClient()) {
            for (int i = 0; i < 10; i++) {
                String i_str = Integer.toString(i);
                System.out.println("[x] Requesting fib(" + i_str + ")");

                String response = fibonacciRpc.call(i_str);
                System.out.println("[.] Got'" + response + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    //请求服务端
    private String call(String message) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();

        //获取队列名称
        String replyQueueName = channel.queueDeclare().getQueue();

        //设置 replyTo 队列和 correlationId请求标识
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        //发送消息至队列
        channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes(StandardCharsets.UTF_8));

        //设置线程等待，每次只接收一个响应结果
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

        //接收服务器返回结果
        String ctag = channel.basicConsume(replyQueueName, true, ((consumerTag, delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response.offer(new String(delivery.getBody(), StandardCharsets.UTF_8));
            }
        }), consumerTag -> {
        });

        String result = response.take();
        channel.basicCancel(ctag);
        return result;
    }

    @Override
    public void close() throws IOException {
        connection.close();
    }
}
